package com.yjd.bi.etl;

import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.log4j.Logger;
import org.springframework.jdbc.core.JdbcTemplate;

import com.yjd.bi.report.enums.BiDataChangeRecordStatusEunm;
import com.yjd.bi.report.enums.BiDataChangeRecordTypeEnum;
import com.yjd.bi.report.enums.BiEtlFromtableTotableIsActivieEnum;
import com.yjd.bi.report.model.BiDatabaseInfoModel;
import com.yjd.bi.report.model.BiEtlFromtableTotableModel;
import com.yjd.bi.report.model.BiTableInfoModel;
import com.yjd.bi.report.service.IBiDatabaseInfoService;
import com.yjd.bi.report.service.IBiEtlFromtableTotableService;
import com.yjd.bi.report.service.IBiTableInfoService;
import com.yjd.bi.report.util.BiUtil;
import com.yjd.comm.annotation.ServiceRemoteProxy;
import com.yjd.comm.base.concurrent.task.ITaskHandler;
import com.yjd.comm.base.concurrent.task.ITaskItem;
import com.yjd.comm.base.concurrent.task.TaskDisposeUtil;
import com.yjd.comm.base.enums.DbWREnums;
import com.yjd.job.BaseJob;
import com.yjd.comm.rpm.RpmServiceKeyEnum;
import com.yjd.comm.util.FrameUtil;
import com.yjd.comm.util.StringUtil;
import com.yjd.comm.vo.IsSuccess;

/**
 * <b>description</b>：将业务库的数据以日志的形式迁移到bi库，然后处理变更日志的job再通过变更日志抓取业务库数据入bi库 <br>
 * <b>time</b>：2016-4-8下午3:47:51 <br>
 * <b>author</b>： ready likun_557@163.com
 */
public class MoveDataAsChangeLogFromBsToBiJob extends BaseJob {

	private Logger logger = Logger
			.getLogger(MoveDataAsChangeLogFromBsToBiJob.class);

	@SuppressWarnings("unchecked")
	@Override
	protected boolean doExecute() throws Exception {
		List<BiEtlFromtableTotableModel> list = this.biEtlFromtableTotableService
				.getModelList(FrameUtil.newHashMap("is_activie",
						BiEtlFromtableTotableIsActivieEnum.YES.getValue()),
						DbWREnums.READ);
		final IsSuccess isSuccess = new IsSuccess();
		TaskDisposeUtil.dispose(true, 10, list,
				new ITaskHandler<BiEtlFromtableTotableModel>() {
					@Override
					public void dispose(
							ITaskItem<BiEtlFromtableTotableModel> item) {
						try {
							moveDataFromBsToBiChangeLog(item.getItem().getId());
						} catch (Exception e) {
							isSuccess.success = false;
							logger.error("全量更新抓取数据日志出错:"
									+ FrameUtil.json(item.getItem()));
							logger.error("全量更新抓取数据日志出错:" + e.getMessage(), e);
						}
					}
				});
		return isSuccess.success;
	}

	/**
	 * 将业务数据以变更日志的方式插入变更日志库
	 * 
	 * @param m
	 * @throws Exception
	 */
	@SuppressWarnings("unchecked")
	public void moveDataFromBsToBiChangeLog(long bi_etl_fromtable_totable_id)
			throws Exception {
		BiEtlFromtableTotableModel m = this.biEtlFromtableTotableService
				.getModelById(bi_etl_fromtable_totable_id, DbWREnums.READ);
		if (m != null) {
			Map<Long, BiTableInfoModel> tableMap = this.biTableInfoService
					.getModelMapByIds(
							FrameUtil.newArrayList(m.getFrom_table_id(),
									m.getTo_table_id()), DbWREnums.READ);
			BiTableInfoModel fromTable = tableMap.get(m.getFrom_table_id());
			BiTableInfoModel toTable = tableMap.get(m.getTo_table_id());
			if (fromTable == null) {
				this.logger.warn("fromTable,id=" + m.getFrom_table_id()
						+ " 为空!");
			} else if (toTable == null) {
				this.logger.warn("toTable,id=" + m.getTo_table_id() + " 为空!");
			} else {
				Map<Long, BiDatabaseInfoModel> dbMap = this.biDatabaseInfoService
						.getModelMapByIds(FrameUtil.newArrayList(
								fromTable.getDatabase_info_id(),
								toTable.getDatabase_info_id()), DbWREnums.READ);
				BiDatabaseInfoModel fromDb = dbMap.get(fromTable
						.getDatabase_info_id());
				BiDatabaseInfoModel toDb = dbMap.get(toTable
						.getDatabase_info_id());
				if (fromDb == null) {
					this.logger.warn("fromDb,id="
							+ fromTable.getDatabase_info_id() + " 为空!");
				} else if (toDb == null) {
					if (toDb == null) {
						this.logger.warn("toDb,id="
								+ toTable.getDatabase_info_id() + " 为空!");
					}
				} else {
					List<Object> dataPkList = getNotDataPkListFromBs(fromDb,
							fromTable, toDb, toTable, m);
					if (dataPkList != null && !dataPkList.isEmpty()) {
						insertToBiChangeLog(dataPkList, fromDb, toDb, toTable,
								m);
						moveDataFromBsToBiChangeLog(bi_etl_fromtable_totable_id);
					}
				}
			}
		}
	}

	public void insertToBiChangeLog(final List<Object> dataPkList,
			BiDatabaseInfoModel fromDb, BiDatabaseInfoModel toDb,
			BiTableInfoModel fromTable, BiEtlFromtableTotableModel m)
			throws Exception {

		if (dataPkList != null && !dataPkList.isEmpty() && fromTable != null) {
			final String etl_all_max_uq_value = dataPkList.get(dataPkList
					.size() - 1) + "";
			final long bi_etl_fromtable_totable_id = m.getId();
			@SuppressWarnings("unchecked")
			final List<Object[]> batchArgs = FrameUtil.newArrayList();
			Long curtime = FrameUtil.getTime();
			for (Object object : dataPkList) {
				int argsIndex = 0;
				Object[] args = new Object[8];
				args[argsIndex++] = fromDb.getId();
				args[argsIndex++] = -1;
				args[argsIndex++] = fromTable.getTable_name();
				args[argsIndex++] = object.toString();
				args[argsIndex++] = BiDataChangeRecordTypeEnum.ADD.getValue();
				args[argsIndex++] = curtime;
				args[argsIndex++] = BiDataChangeRecordStatusEunm.WAIT_DISPOSE
						.getValue();
				args[argsIndex++] = curtime;
				batchArgs.add(args);
			}
			BiUtil.execute(toDb, new BiUtil.ITransExecute<Boolean>() {
				@Override
				public Boolean execute(JdbcTemplate jt) throws Exception {
					jt.batchUpdate(
							"INSERT INTO yjd_bi_data_change_record "
									+ "(database_info_id, data_change_record_id, table_name, ref_id, type, create_time, status, dispose_time)"
									+ " VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
							batchArgs);
					jt.update(
							"update yjd_bi_etl_fromtable_totable set etl_all_max_uq_value = ? where id = ?",
							etl_all_max_uq_value, bi_etl_fromtable_totable_id);
					return true;
				}
			});
		}
	}

	public void insertToBiData(final List<Map<String, Object>> rows,
			BiDatabaseInfoModel dbm, BiTableInfoModel table,
			BiEtlFromtableTotableModel m) throws Exception {
		if (rows != null && !rows.isEmpty() && table != null) {
			StringBuilder colSql = new StringBuilder("");
			StringBuilder colParam = new StringBuilder("");
			Set<String> columnList = rows.get(0).keySet();
			for (String column : columnList) {
				colSql.append(column).append(",");
				colParam.append("?").append(",");
			}
			final String sql = String.format("insert into %s (%s) values (%s)",
					table.getTable_name(),
					colSql.substring(0, colSql.length() - 1),
					colParam.substring(0, colParam.length() - 1));
			final String etl_all_max_uq_value = rows.get(rows.size() - 1).get(
					table.getPk_field_name())
					+ "";
			final long bi_etl_fromtable_totable = m.getId();
			@SuppressWarnings("unchecked")
			final List<Object[]> batchArgs = FrameUtil.newArrayList();
			for (Map<String, Object> map : rows) {
				batchArgs.add(map.values().toArray());
			}
			BiUtil.execute(dbm, new BiUtil.ITransExecute<Boolean>() {
				@Override
				public Boolean execute(JdbcTemplate jt) throws Exception {
					jt.batchUpdate(sql, batchArgs);
					jt.update(
							"update yjd_bi_etl_fromtable_totable set etl_all_max_uq_value = ? where id = ?",
							etl_all_max_uq_value, bi_etl_fromtable_totable);
					return true;
				}
			});
		}
	}

	/**
	 * 获取还未从业务库移到bi库的数据主键列表
	 * 
	 * @param fromDb
	 * @param fromTable
	 * @param toDb
	 * @param toTable
	 * @param m
	 * @return
	 */
	public List<Object> getNotDataPkListFromBs(BiDatabaseInfoModel fromDb,
			BiTableInfoModel fromTable, BiDatabaseInfoModel toDb,
			BiTableInfoModel toTable, BiEtlFromtableTotableModel m) {
		if (fromDb != null && fromTable != null && toDb != null
				&& toTable != null) {
			StringBuilder fromSql = new StringBuilder("select "
					+ fromTable.getPk_field_name() + " from ");
			fromSql.append(fromTable.getTable_name());
			if (StringUtil.isNotEmpty(m.getEtl_all_max_uq_value())) {
				fromSql.append(" where ").append(fromTable.getPk_field_name())
						.append(" > ").append(m.getEtl_all_max_uq_value());
			}
			fromSql.append(" order by ").append(fromTable.getPk_field_name())
					.append(" asc limit 0,500");
			List<Object> fromTableData = BiUtil.getJdbcTemplate(fromDb)
					.queryForList(fromSql.toString(), Object.class);
			return fromTableData;
		}
		return null;
	}

	@Override
	protected String getJobType() {
		return MoveDataAsChangeLogFromBsToBiJob.class.getName();
	}

	/**
	 * 数据库信息服务
	 */
	@ServiceRemoteProxy(serviceUrlResouceKey = RpmServiceKeyEnum.RPM_BI_SERVICE_KEY)
	private IBiDatabaseInfoService biDatabaseInfoService;
	/**
	 * 数据抽取中原表和目标表关联信息服务
	 */
	@ServiceRemoteProxy(serviceUrlResouceKey = RpmServiceKeyEnum.RPM_BI_SERVICE_KEY)
	private IBiEtlFromtableTotableService biEtlFromtableTotableService;

	/**
	 * 表信息服务
	 */
	@ServiceRemoteProxy(serviceUrlResouceKey = RpmServiceKeyEnum.RPM_BI_SERVICE_KEY)
	private IBiTableInfoService biTableInfoService;

}
